package edu.zju.gis.dbfg.model.statistic;

import com.mysql.cj.protocol.a.NativeConstants;
import edu.zju.gis.dbfg.common.exception.ModelFailedException;
import edu.zju.gis.dbfg.model.PyramidConfig;
import edu.zju.gis.dbfg.model.datasource.DataReaderRDD;
import edu.zju.gis.dbfg.model.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.*;
import org.geotools.referencing.CRS;
import org.locationtech.jts.geom.Geometry;
import org.opengis.referencing.crs.CoordinateReferenceSystem;
import org.opengis.referencing.operation.MathTransform;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;
import scala.reflect.internal.Trees;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.*;

import static edu.zju.gis.dbfg.model.base.Constant.*;

public class LUAreaStatistic2D {
    private final static Logger logger = LoggerFactory.getLogger(LUAreaStatistic2D.class);
    public static void main(String[] args) throws Exception {
        logger.info("Setup Spark Context");
        SparkConf conf = new SparkConf();
        conf.setMaster("local[1]");
        conf.setAppName("lalala");
        JavaSparkContext jsc = new JavaSparkContext(conf);

        logger.info("Setup arguments");
        LUAreaStatisticArgs2D mArgs = LUAreaStatisticArgs2D.initArgs(args);
        if (mArgs == null) {
            String msg = "init args failed, exit";
            logger.error(msg);
            throw new ModelFailedException(LUAreaStatistic.class, "main()", msg, args);
        }

        CoordinateReferenceSystem crs1 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs1()));
        CoordinateReferenceSystem crs2 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs2()));
        CoordinateReferenceSystem crs3 = CRS.parseWKT(CrsPick2.pickByName(mArgs.getCrs3()));
        CoordinateReferenceSystem crs4 = CRS.parseWKT(CrsPick.pickByName(mArgs.getCrs4()));

        String output = mArgs.getOutput();
        String input1 = mArgs.getInput1();
        String type1 = mArgs.getType1();
        String input2 = mArgs.getInput2();
        String type2 = mArgs.getType2();
        String zldwdm2 = mArgs.getZldwdmIndex2();
        String dlbm2 = mArgs.getDlbmIndex2();
        String area2 = mArgs.getAreaIndex2();
        String jarea2 = mArgs.getJareaIndex2();
        String kcdlbm2 = mArgs.getKcdlbmIndex2();
        String kcdlxs2 = mArgs.getKcdlxsIndex2();
        String tbbh2 = mArgs.getTbbhIndex2();
        String tbybh2 = mArgs.getTbybhIndex2();
        String xzdwmj2 = mArgs.getXzdwmjIndex2();
        String lxdwmj2 = mArgs.getLxdwmjIndex2();
        String scale2 = mArgs.getScaleIndex2();


        String input3 = mArgs.getInput3();
        String type3 = mArgs.getType3();
        String dlbm3 = mArgs.getDlbmIndex3();
        String kctbdwdm13 = mArgs.getKctbdwdm1Index3();
        String kctbdwdm23 = mArgs.getKctbdwdm2Index3();
        String kctbbh13 = mArgs.getKctbbh1Index3();
        String kctbbh23 = mArgs.getKctbbh2Index3();
        String kcbl3 = mArgs.getKcblIndex3();
        String cd3 = mArgs.getCdIndex3();
        String kd3 = mArgs.getKdIndex3();


        String input4 = mArgs.getInput4();
        String type4 = mArgs.getType4();
        String dlbm4 = mArgs.getDlbmIndex4();
        String zldwdm4 = mArgs.getZldwdmIndex4();
        String zltbbh4 = mArgs.getZltbbhIndex4();
        String mj4 = mArgs.getMjIndex4();



        Map<String,GeoItem> results= areaStatistic2d(jsc, crs1, crs2, crs3, crs4,  input1, type1, input2, type2, zldwdm2, dlbm2, area2, jarea2, kcdlbm2, kcdlxs2, tbbh2, tbybh2, xzdwmj2, lxdwmj2, scale2, input3, type3, dlbm3, kctbdwdm13, kctbdwdm23, kctbbh13, kctbbh23, kcbl3, cd3, kd3, input4, type4, dlbm4, zldwdm4, zltbbh4, mj4);

        File file = new File(output);
        if (file.exists()) {
            file.delete();
        }
        FileWriter writer = new FileWriter(file);
        WriteToFile.writeToFile(writer,results);
        writer.close();

    }

    public static Map<String, GeoItem> areaStatistic2d(JavaSparkContext jsc,CoordinateReferenceSystem crs1,CoordinateReferenceSystem crs2,CoordinateReferenceSystem crs3,CoordinateReferenceSystem crs4,String input1,String type1,String input2,String type2,String zldwdm2,String dlbm2,String area2,String jarea2,String kcdlbm2,String kcdlxs2,String tbbh2,String tbybh2,String xzdwmj2,String lxdwmj2,String scale2,String input3,String type3,String dlbm3,String kctbdwdm13,String kctbdwdm23,String kctbbh13,String kctbbh23,String kcbl3,String cd3,String kd3,String input4,String type4,String dlbm4,String zldwdm4,String zltbbh4,String mj4) throws Exception{
        CoordinateReferenceSystem targetCRS = DEFAULT_CRS();
        PyramidConfig targetPyramidConfig = PYRAMID_CONFIG_4326();//gewang
        //PyramidConfig targetPyramidConfig = DEFAULT_PYRAMID_CONFIG();
        MathTransform transform1 = null;
        MathTransform transform2 = null;
        MathTransform transform3 = null;
        MathTransform transform4 = null;
        if (!crs1.equals(targetCRS)) {
            transform1 = CRS.findMathTransform(crs1, targetCRS);
        }
        if (!crs2.equals(targetCRS)) {
            transform2 = CRS.findMathTransform(crs2, targetCRS);
        }
        if (!crs3.equals(targetCRS)) {
            transform1 = CRS.findMathTransform(crs3, targetCRS);
        }
        if (!crs4.equals(targetCRS)) {
            transform2 = CRS.findMathTransform(crs4, targetCRS);
        }



        Map<String, Object> attrs1 = new HashMap<>();
        JavaRDD<GeoItem> gi1 = DataReaderRDD.readToRDD(type1, jsc, input1, "1", attrs1, transform1);
        JavaPairRDD<String, GeoItem> gi1Pair = DataReaderRDD.mapToPair(gi1, targetPyramidConfig);



        Map<String, Object> attrs2 = new HashMap<>();
        attrs2.put("DLBM", dlbm2);
        attrs2.put("ZLDWDM", zldwdm2);
        attrs2.put("TBMJ", area2);
        attrs2.put("TBDLMJ", jarea2);
        attrs2.put("KCDLBM", kcdlbm2);
        attrs2.put("KCDLXS", kcdlxs2);
        attrs2.put("TBBH", tbbh2);
        attrs2.put("TBYBH", tbybh2);
        attrs2.put("XZDWMJ", xzdwmj2);
        attrs2.put("LXDWMJ", lxdwmj2);

        JavaPairRDD<String, GeoItem> gi2Pair = null;

        if (type2.equals("ES")) {
            gi1Pair.cache();
            List<TileID> tileIDs = gi1Pair.map(t -> TileID.fromString(t._1)).distinct().collect();
            gi2Pair = DataReaderRDD.readToPairRDD(type2, jsc, input2, attrs2, tileIDs, targetPyramidConfig);
//            JavaRDD<GeoItem>  gi2 = DataReaderRDD.readToRDD2(type2, jsc, input2, attrs2, null, null, transform2);
//            gi2Pair = DataReaderRDD.mapToPair(gi2, targetPyramidConfig);
        } else {
            JavaRDD<GeoItem> gi2 = DataReaderRDD.readToRDD(type2, jsc, input2, "1", attrs2, transform2);
            gi2Pair = DataReaderRDD.mapToPair(gi2, targetPyramidConfig);
        }

        JavaPairRDD<String, Tuple2<GeoItem, GeoItem>> joined2 = gi1Pair.join(gi2Pair);
        JavaPairRDD<String, GeoItem> dltbMdlRDD = joined2.mapToPair(new PairFunction<Tuple2<String, Tuple2<GeoItem, GeoItem>>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, Tuple2<GeoItem, GeoItem>> input) throws Exception {
                GeoItem gi2 = input._2._2;
                GeoItem gi1 = input._2._1;
                Geometry g1 = input._2._1.getGeometry();
                Geometry g2 = input._2._2.getGeometry();
                if (!g1.intersects(g2)) {
                    return new Tuple2<>("EMPTY", null);
                }

                Geometry intersection = g1.intersection(g2);
                Double iarea = intersection.getArea();
                //Double oarea = g2.getArea();
                String dlbm = gi2.getProp().get("DLBM").toString().replace("\"", "").replace("\40", "");
                String zldwdm = String.valueOf(gi2.getProp().get("ZLDWDM")).replace("\"", "").replace("\40", "").substring(0,12);
                Double tbdlmj = Double.valueOf(gi2.getProp().get("TBDLMJ").toString());
                String kcdlbm = String.valueOf(gi2.getProp().get("KCDLBM")).replace("\"", "").replace("\40", "");
                Double tbmj = Double.valueOf(gi2.getProp().get("TBMJ").toString());
                Double kcdlxs = Double.valueOf(gi2.getProp().get("KCDLXS").toString());
                String tbbh = String.valueOf(gi2.getProp().get("TBBH")).replace("\"", "").replace("\40", "");
                //String tbybh = String.valueOf(gi2.getProp().get("TBYBH")).replace("\"", "");
                Double xzdwmj = Double.valueOf(gi2.getProp().get("XZDWMJ").toString());
                Double lxdwmj = Double.valueOf(gi2.getProp().get("LXDWMJ").toString());

                Double ytbmj = Double.valueOf(gi2.getProp().get("YTBMJ").toString());
                Double zero = 0.0;




                tbmj = iarea / ytbmj * tbmj;
                String id = input._1+'|'+tbbh + "##" + zldwdm;
                GeoItem result = new GeoItem();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("ZLDWDM", zldwdm);
                props.put("DLBM", dlbm);
                props.put("KCDLBM", kcdlbm);
                props.put("TBMJ", tbmj);//tbmj
                props.put("TBDLMJ", zero);
                props.put("KCDLXS", kcdlxs);
                props.put("KCDLMJ", zero);
                props.put("TBYBH", tbbh);
                props.put("TBBH", tbbh);
                props.put("XZDWMJ", zero);
                props.put("LXDWMJ", zero);

                result.setProp(props);
                result.setGeometry(intersection);

                if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
                {
                    System.out.println(dlbm+","+zldwdm+","+iarea+","+xzdwmj+","+lxdwmj+","+tbdlmj);
                }

                return new Tuple2<>(id, result);
            }

        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });
        dltbMdlRDD.cache();
        //WriteWKT.writeToFile(dltbMdlRDD, "G:\\GIT\\TestData\\wkt\\dltbMdlRDD.wkt");



        Map<String, Object> attrs3 = new HashMap<>();
        attrs3.put("DLBM", dlbm3);
        attrs3.put("KCTBDWDM1", kctbdwdm13);
        attrs3.put("KCTBDWDM2", kctbdwdm23);
        attrs3.put("KCTBBH1", kctbbh13);
        attrs3.put("KCTBBH2", kctbbh23);
        attrs3.put("KCBL", kcbl3);
        attrs3.put("CD", cd3);
        attrs3.put("KD", kd3);

        JavaPairRDD<String, GeoItem> gi3Pair = null;

        if (type3.equals("ES")) {
            List<TileID> tileIDs = gi1Pair.map(t -> TileID.fromString(t._1)).distinct().collect();
            gi3Pair = DataReaderRDD.readToPairRDD(type3, jsc, input3, attrs3, tileIDs, targetPyramidConfig);
        } else {
            JavaRDD<GeoItem> gi3 = DataReaderRDD.readToRDD(type3, jsc, input3, "1", attrs3, transform3);
            gi3Pair = DataReaderRDD.mapToPair(gi3, targetPyramidConfig);
        }

        JavaPairRDD<String, Tuple2<GeoItem, GeoItem>> joined3 = gi1Pair.join(gi3Pair);
        JavaPairRDD<String, GeoItem> dltbXzdwRDD1 = joined3.mapToPair(new PairFunction<Tuple2<String, Tuple2<GeoItem, GeoItem>>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, Tuple2<GeoItem, GeoItem>> input) throws Exception {
                GeoItem gi2 = input._2._2;
                GeoItem gi1 = input._2._1;
                Geometry g1 = input._2._1.getGeometry();
                Geometry g2 = input._2._2.getGeometry();
                if (!g1.intersects(g2)) {
                    return new Tuple2<>("EMPTY", null);
                }
                Geometry intersection = g1.intersection(g2);
                Double ilength = intersection.getLength();
                //Double olength = g2.getLength();
                String dlbm = gi2.getProp().get("DLBM").toString().replace("\"", "").replace("\40", "");
                String kctbdwdm1 = String.valueOf(gi2.getProp().get("KCTBDWDM1")).replace("\"", "").replace("\40", "").substring(0,12);
                String kctbdwdm2 = String.valueOf(gi2.getProp().get("KCTBDWDM2")).replace("\"", "").replace("\40", "");
                if(kctbdwdm2.length()>12){
                    kctbdwdm2=kctbdwdm2.substring(0,12);
                }
                String kctbbh1 = String.valueOf(gi2.getProp().get("KCTBBH1")).replace("\"", "").replace("\40", "");
                String kctbbh2 = String.valueOf(gi2.getProp().get("KCTBBH2")).replace("\"", "").replace("\40", "");
                Double kcbl = Double.valueOf(gi2.getProp().get("KCBL").toString());
                Double cd = Double.valueOf(gi2.getProp().get("CD").toString());
                Double kd = Double.valueOf(gi2.getProp().get("KD").toString());
                Double ylength = Double.valueOf(gi2.getProp().get("YLENGTH").toString());
                cd = cd * ilength / ylength;
                String id =input._1+"|" + kctbbh1 + "|" + kctbdwdm1;
                GeoItem result = new GeoItem();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("TILEID", input._1);
                props.put("DLBM", dlbm);
                props.put("KCTBDWDM1", kctbdwdm1);
                props.put("KCTBDWDM2", kctbdwdm2);
                props.put("KCTBBH1", kctbbh1);
                props.put("KCTBBH2", kctbbh2);
                props.put("KCBL", kcbl);
                props.put("CD", cd);
                props.put("KD", kd);
                props.put("TBMJ", cd*kd);
                result.setProp(props);
                result.setGeometry(intersection);
                return new Tuple2<>(id, result);
            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });
        dltbXzdwRDD1.cache();
        //WriteWKT.writeToFile(dltbXzdwRDD1, "G:\\GIT\\TestData\\wkt\\dltbXzdwRDD.wkt");

        JavaPairRDD<String, GeoItem> dltbXzdwRDD2=dltbXzdwRDD1.mapToPair(mdl->new Tuple2<>(mdl._2.getProp().get("TILEID")+"|"+mdl._2.getProp().get("KCTBBH2")+"|"+mdl._2.getProp().get("KCTBDWDM2"),mdl._2));
        dltbXzdwRDD2.cache();
        Map<String, Object> attrs4 = new HashMap<>();
        attrs4.put("DLBM", dlbm4);
        attrs4.put("ZLDWDM", zldwdm4);
        attrs4.put("ZLTBBH", zltbbh4);
        attrs4.put("MJ", mj4);

        JavaPairRDD<String, GeoItem> gi4Pair = null;

        if (type4.equals("ES")) {
            List<TileID> tileIDs = gi1Pair.map(t -> TileID.fromString(t._1)).distinct().collect();
            gi4Pair = DataReaderRDD.readToPairRDD(type4, jsc, input4, attrs4, tileIDs, targetPyramidConfig);
        } else {
            JavaRDD<GeoItem> gi4 = DataReaderRDD.readToRDD(type4, jsc, input4, "1", attrs4, transform4);
            gi4Pair = DataReaderRDD.mapToPair(gi4, targetPyramidConfig);
        }
        JavaPairRDD<String, Tuple2<GeoItem, GeoItem>> joined4 = gi1Pair.join(gi4Pair);
        JavaPairRDD<String, GeoItem> dltbLxdwRDD = joined4.mapToPair(new PairFunction<Tuple2<String, Tuple2<GeoItem, GeoItem>>, String, GeoItem>() {
            @Override
            public Tuple2<String, GeoItem> call(Tuple2<String, Tuple2<GeoItem, GeoItem>> input) throws Exception {
                GeoItem gi2 = input._2._2;
                GeoItem gi1 = input._2._1;
                Geometry g1 = input._2._1.getGeometry();
                Geometry g2 = input._2._2.getGeometry();
                if (!g1.intersects(g2)) {
                    return new Tuple2<>("EMPTY", null);
                }
                String dlbm = gi2.getProp().get("DLBM").toString().replace("\"", "").replace("\40", "");
                String zldwdm = String.valueOf(gi2.getProp().get("ZLDWDM")).replace("\"", "").replace("\40", "").substring(0,12);
                String zltbbh = String.valueOf(gi2.getProp().get("ZLTBBH")).replace("\"", "").replace("\40", "");
                Double mj = Double.valueOf(gi2.getProp().get("MJ").toString());
                String id = input._1+'|'+zltbbh + "|" + zldwdm;
                GeoItem result = new GeoItem();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("DLBM", dlbm);
                props.put("ZLDWDM", zldwdm);
                props.put("ZLTBBH", zltbbh);
                props.put("MJ", mj);

                result.setProp(props);
                result.setGeometry(g2);
                return new Tuple2<>(input._1, result);

            }
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

//        dltbLxdwRDD.cache();
//        WriteWKT.writeToFile(dltbLxdwRDD, "G:\\GIT\\TestData\\wkt\\dltbLxdwRDD.wkt");
        JavaPairRDD<String, Iterable<GeoItem>> dltbXzdwRDDlist1=dltbXzdwRDD1.groupByKey();
        JavaPairRDD<String, Tuple2<String,GeoItem>>  dltbMdlRDDtemp1=dltbMdlRDD.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" + mdl._2.getProp().get("TBBH") + "|" + mdl._2.getProp().get("ZLDWDM"), mdl));
        dltbMdlRDDtemp1.cache();
        Map<String, Tuple2<String,GeoItem>> test=dltbMdlRDDtemp1.collectAsMap();
        JavaPairRDD<String, Tuple2<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>>>  dltbMdlRDDxz1=dltbMdlRDDtemp1.leftOuterJoin(dltbXzdwRDDlist1);
        dltbMdlRDDxz1.cache();
        Map<String, Tuple2<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>>> test2=dltbMdlRDDxz1.collectAsMap();
        JavaPairRDD<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>> dltbMdlRDDxz1map=dltbMdlRDDxz1.mapToPair(Tuple2::_2);
        dltbMdlRDDxz1map.cache();
        Map<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>> test3=dltbMdlRDDxz1map.collectAsMap();
        JavaPairRDD<Tuple2<String,GeoItem>,Iterable<Optional<Iterable<GeoItem>>>> dltbMdlRDDxz1mapi=dltbMdlRDDxz1map.groupByKey();
        dltbMdlRDDxz1mapi.cache();
        Map<Tuple2<String,GeoItem>,Iterable<Optional<Iterable<GeoItem>>>> test4=dltbMdlRDDxz1mapi.collectAsMap();
        JavaPairRDD<String, GeoItem> dltbMdlRDDxz1re=dltbMdlRDDxz1mapi.mapToPair(tp -> {
            GeoItem geoItem1 = tp._1._2;
            String fid1 = tp._1._1;
            Double xzdwmj = Double.valueOf(geoItem1.getProp().get("XZDWMJ").toString());
            for (Optional<Iterable<GeoItem>> optional : tp._2()) {
                if (optional.isPresent()) {
                    Iterable<GeoItem> iterable = optional.get();
                    for(GeoItem geoItem :iterable){
                        Geometry geo1buffer=geoItem1.getGeometry().buffer(0.02,30);
                        if (geo1buffer.contains(geoItem.getGeometry())) {
                            Double cd = Double.valueOf(geoItem.getProp().get("CD").toString());
                            Double kd = Double.valueOf(geoItem.getProp().get("KD").toString());
                            Double kcbl = Double.valueOf(geoItem.getProp().get("KCBL").toString());
                            xzdwmj += cd * kd * kcbl;
                        }
                    }
                }
            }
            Map<String, Object> props = geoItem1.getProp();
            props.replace("XZDWMJ", xzdwmj);
            geoItem1.setProp(props);
            String dlbm=geoItem1.getProp().get("DLBM").toString();
            String zldwdm=geoItem1.getProp().get("ZLDWDM").toString();
            Double tbmj=Double.parseDouble(geoItem1.getProp().get("TBMJ").toString());
            if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
            {
                System.out.println(dlbm+","+zldwdm+","+xzdwmj+","+tbmj);
            }
            return new Tuple2<>(fid1, geoItem1);
        });

        JavaPairRDD<String, Iterable<GeoItem>> dltbXzdwRDDlist2=dltbXzdwRDD2.groupByKey();
        JavaPairRDD<String, Tuple2<String,GeoItem>>  dltbMdlRDDtemp2=dltbMdlRDDxz1re.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" + mdl._2.getProp().get("TBBH") + "|" + mdl._2.getProp().get("ZLDWDM"), mdl));

        JavaPairRDD<String, Tuple2<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>>>  dltbMdlRDDxz2=dltbMdlRDDtemp2.leftOuterJoin(dltbXzdwRDDlist2);
        JavaPairRDD<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>> dltbMdlRDDxz2map=dltbMdlRDDxz2.mapToPair(Tuple2::_2);
        JavaPairRDD<Tuple2<String,GeoItem>,Iterable<Optional<Iterable<GeoItem>>>> dltbMdlRDDxz2mapi=dltbMdlRDDxz2map.groupByKey();
        JavaPairRDD<String, GeoItem> dltbMdlRDDxz2re=dltbMdlRDDxz2mapi.mapToPair(tp -> {
            GeoItem geoItem1 = tp._1._2;
            String fid1 = tp._1._1;
            Double xzdwmj = Double.valueOf(geoItem1.getProp().get("XZDWMJ").toString());

            for (Optional<Iterable<GeoItem>> optional : tp._2()) {
                if (optional.isPresent()) {

                    Iterable<GeoItem> iterable = optional.get();
                    //String tbbh=geoItem2.getProp().get("KCTBBH2").toString();
                    for(GeoItem geoItem :iterable){
                        Geometry geo1buffer=geoItem1.getGeometry().buffer(0.02,30);
                        if (geo1buffer.contains(geoItem.getGeometry())) {
                            Double cd = Double.valueOf(geoItem.getProp().get("CD").toString());
                            Double kd = Double.valueOf(geoItem.getProp().get("KD").toString());
                            Double kcbl = Double.valueOf(geoItem.getProp().get("KCBL").toString());
                            xzdwmj += cd * kd * kcbl ;
                        }
                    }


                }
            }
            Map<String, Object> props = geoItem1.getProp();
            props.replace("XZDWMJ", xzdwmj);
            geoItem1.setProp(props);
            String dlbm=geoItem1.getProp().get("DLBM").toString();
            String zldwdm=geoItem1.getProp().get("ZLDWDM").toString();
            Double tbmj=Double.parseDouble(geoItem1.getProp().get("TBMJ").toString());
//                    if(dlbm.equals("203")&&(zldwdm.equals("330802009007")||zldwdm.equals("3308020090070000000")))
//                    {
//                         System.out.println(dlbm+","+zldwdm+","+xzdwmj+","+tbmj);
//                    }
            if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
            {
                System.out.println(dlbm+","+zldwdm+","+xzdwmj+","+tbmj);
            }
            return new Tuple2<>(fid1, geoItem1);
        });


//        JavaPairRDD<String, GeoItem>  dltbMdlRDDxz = dltbMdlRDD.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" + mdl._2.getProp().get("TBBH") + "|" + mdl._2.getProp().get("ZLDWDM"), mdl))
//                .leftOuterJoin(dltbXzdwRDD.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" +mdl._2.getProp().get("KCTBBH1") + "|" + mdl._2.getProp().get("KCTBDWDM1"), mdl)))
//                .mapToPair(Tuple2::_2)
//                .groupByKey()
//                .mapToPair(tp -> {
//                    GeoItem geoItem1 = tp._1._2;
//                    String fid1 = tp._1._1;
//                    Double xzdwmj = Double.valueOf(geoItem1.getProp().get("XZDWMJ").toString());
//
//                    for (Optional<Tuple2<String, GeoItem>> optional : tp._2()) {
//                        if (optional.isPresent()) {
//                            GeoItem geoItem2 = optional.get()._2;
//                            Geometry geo1buffer=geoItem1.getGeometry().buffer(0.02,30);
//                            if (geo1buffer.contains(geoItem2.getGeometry())) {
//                                Double cd = Double.valueOf(geoItem2.getProp().get("CD").toString());
//                                Double kd = Double.valueOf(geoItem2.getProp().get("KD").toString());
//                                Double kcbl = Double.valueOf(geoItem2.getProp().get("KCBL").toString());
//                                xzdwmj += cd * kd * kcbl;
//                            }
//                        }
//                    }
//                    Map<String, Object> props = geoItem1.getProp();
//                    props.replace("XZDWMJ", xzdwmj);
//                    geoItem1.setProp(props);
//                    String dlbm=geoItem1.getProp().get("DLBM").toString();
//                    String zldwdm=geoItem1.getProp().get("ZLDWDM").toString();
//                    Double tbmj=Double.parseDouble(geoItem1.getProp().get("TBMJ").toString());
//                    if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
//                    {
//                        System.out.println(dlbm+","+zldwdm+","+xzdwmj+","+tbmj);
//                    }
//                    return new Tuple2<>(fid1, geoItem1);
//
////                    while (tp._2.iterator().hasNext()){
////                       Tuple2<String,GeoItem> tu = tp._2.iterator().next();
////                       tp._2.
////                    }
//
//                })
//                .mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" +mdl._2.getProp().get("TBBH") + "|" + mdl._2.getProp().get("ZLDWDM"), mdl))
//                .leftOuterJoin(dltbXzdwRDD.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" +mdl._2.getProp().get("KCTBBH2") + "|" + mdl._2.getProp().get("KCTBDWDM2"), mdl)))
//                .mapToPair(Tuple2::_2)
//                .groupByKey()
//                .mapToPair(tp -> {
//                    GeoItem geoItem1 = tp._1._2;
//                    String fid1 = tp._1._1;
//                    Double xzdwmj = Double.valueOf(geoItem1.getProp().get("XZDWMJ").toString());
//
//                    for (Optional<Tuple2<String, GeoItem>> optional : tp._2()) {
//                        if (optional.isPresent()) {
//
//                            GeoItem geoItem2 = optional.get()._2;
//                            //String tbbh=geoItem2.getProp().get("KCTBBH2").toString();
//
//                            Geometry geo1buffer=geoItem1.getGeometry().buffer(0.02,30);
//                            if (geo1buffer.contains(geoItem2.getGeometry())) {
//                                Double cd = Double.valueOf(geoItem2.getProp().get("CD").toString());
//                                Double kd = Double.valueOf(geoItem2.getProp().get("KD").toString());
//                                Double kcbl = Double.valueOf(geoItem2.getProp().get("KCBL").toString());
//                                xzdwmj += cd * kd * kcbl ;
//                            }
//
//                        }
//                    }
//                    Map<String, Object> props = geoItem1.getProp();
//                    props.replace("XZDWMJ", xzdwmj);
//                    geoItem1.setProp(props);
//                    String dlbm=geoItem1.getProp().get("DLBM").toString();
//                    String zldwdm=geoItem1.getProp().get("ZLDWDM").toString();
//                    Double tbmj=Double.parseDouble(geoItem1.getProp().get("TBMJ").toString());
////                    if(dlbm.equals("203")&&(zldwdm.equals("330802009007")||zldwdm.equals("3308020090070000000")))
////                    {
////                         System.out.println(dlbm+","+zldwdm+","+xzdwmj+","+tbmj);
////                    }
//                    if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
//                    {
//                        System.out.println(dlbm+","+zldwdm+","+xzdwmj+","+tbmj);
//                    }
//                    return new Tuple2<>(fid1, geoItem1);
//                });

        JavaPairRDD<String, Iterable<GeoItem>> dltbLxdwRDDlist=dltbLxdwRDD.groupByKey();
        JavaPairRDD<String, Tuple2<String,GeoItem>>  dltbMdlRDDtemp3=dltbMdlRDDxz2re.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" + mdl._2.getProp().get("TBBH") + "|" + mdl._2.getProp().get("ZLDWDM"), mdl));
        JavaPairRDD<String, Tuple2<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>>>   dltbMdlRDDlx=dltbMdlRDDtemp3.leftOuterJoin(dltbLxdwRDDlist);

        JavaPairRDD<Tuple2<String,GeoItem>,Optional<Iterable<GeoItem>>> dltbMdlRDDlxmap=dltbMdlRDDlx.mapToPair(Tuple2::_2);
        JavaPairRDD<Tuple2<String,GeoItem>,Iterable<Optional<Iterable<GeoItem>>>> dltbMdlRDDlxmapi=dltbMdlRDDlxmap.groupByKey();

        JavaPairRDD<String, GeoItem>  dltbMdlRDDlxre=dltbMdlRDDlxmapi.mapToPair(tp -> {
            GeoItem geoItem1 = tp._1._2;
            String fid1 = tp._1._1;
            Double lxdwmj = Double.valueOf(geoItem1.getProp().get("LXDWMJ").toString());

            for (Optional<Iterable<GeoItem>> optional : tp._2()) {
                if (optional.isPresent()) {
                    Iterable<GeoItem> iterable = optional.get();
                    for(GeoItem geoItem :iterable){
                        Geometry geo1buffer=geoItem1.getGeometry().buffer(0.02,30);
                        if (geo1buffer.contains(geoItem.getGeometry())) {
                            Double mj = Double.valueOf(geoItem.getProp().get("MJ").toString());
                            lxdwmj += mj;
                        }
                    }
                }
            }
            Map<String, Object> props = geoItem1.getProp();
            props.replace("LXDWMJ", lxdwmj);
            geoItem1.setProp(props);
            String dlbm=geoItem1.getProp().get("DLBM").toString();
            String zldwdm=geoItem1.getProp().get("ZLDWDM").toString();
            Double tbmj=Double.parseDouble(geoItem1.getProp().get("TBMJ").toString());
            if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
            {
                System.out.println(dlbm+","+zldwdm+","+lxdwmj+","+tbmj);
            }
            return new Tuple2<>(fid1, geoItem1);
        });


//        JavaPairRDD<String, GeoItem>  dltbMdlRDDlx = dltbMdlRDDxz2re.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" +mdl._2.getProp().get("TBBH") + "|" + mdl._2.getProp().get("ZLDWDM"), mdl))
//                .leftOuterJoin(dltbLxdwRDD.mapToPair(mdl -> new Tuple2<>(mdl._1 + "|" +mdl._2.getProp().get("ZLTBBH") + "|" + mdl._2.getProp().get("ZLDWDM"), mdl)))
//                .mapToPair(Tuple2::_2)
//                .groupByKey()
//                .mapToPair(tp -> {
//                    GeoItem geoItem1 = tp._1._2;
//                    String fid1 = tp._1._1;
//                    Double lxdwmj = Double.valueOf(geoItem1.getProp().get("LXDWMJ").toString());
//
//                    for (Optional<Tuple2<String, GeoItem>> optional : tp._2()) {
//                        if (optional.isPresent()) {
//                            GeoItem geoItem2 = optional.get()._2;
//                            Geometry geo1buffer=geoItem1.getGeometry().buffer(0.02,30);
//                            if (geo1buffer.contains(geoItem2.getGeometry())) {
//                                Double mj = Double.valueOf(geoItem2.getProp().get("MJ").toString());
//                                lxdwmj += mj;
//                            }
//
//                        }
//                    }
//                    Map<String, Object> props = geoItem1.getProp();
//                    props.replace("LXDWMJ", lxdwmj);
//                    geoItem1.setProp(props);
//                    String dlbm=geoItem1.getProp().get("DLBM").toString();
//                    String zldwdm=geoItem1.getProp().get("ZLDWDM").toString();
//                    Double tbmj=Double.parseDouble(geoItem1.getProp().get("TBMJ").toString());
//                    if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
//                    {
//                        System.out.println(dlbm+","+zldwdm+","+lxdwmj+","+tbmj);
//                    }
//                    return new Tuple2<>(fid1, geoItem1);
//                });

        JavaPairRDD<String, GeoItem> dltbMdlRDDall = dltbMdlRDDlxre.mapToPair(mdl -> {
            Map<String, Object> props = mdl._2.getProp();
            if (props.get("KCDLXS") == ""||props.get("KCDLXS") == null ||props.get("KCDLXS").toString()=="0.0") {// 如果没有扣除地类系数，则将TKMJ赋为0
            } else {
                double tkxs = Double.parseDouble(props.get("KCDLXS").toString());
                if (tkxs > 1)
                    tkxs = tkxs / 100;
                Double kcdlmj = (Double.parseDouble(props.get("TBMJ").toString()) - Double.parseDouble(props.get("XZDWMJ").toString()) - Double.parseDouble(props.get("LXDWMJ").toString())) * tkxs;

                props.replace("KCDLMJ", kcdlmj);
                mdl._2.setProp(props);

            }
            Double tbdlmj = Double.parseDouble(props.get("TBMJ").toString()) - Double.parseDouble(props.get("XZDWMJ").toString()) - Double.parseDouble(props.get("LXDWMJ").toString()) - Double.parseDouble(props.get("KCDLMJ").toString());
            props.replace("TBDLMJ", tbdlmj);
            mdl._2.setProp(props);

            String dlbm=props.get("DLBM").toString();
            String zldwdm=props.get("ZLDWDM").toString();
            Double tbmj=Double.parseDouble(props.get("TBMJ").toString());
            Double xzdwmj=Double.parseDouble(props.get("XZDWMJ").toString());
            Double lxdwmj=Double.parseDouble(props.get("LXDWMJ").toString());
            Double kcdlmj=Double.parseDouble(props.get("KCDLMJ").toString());
            if(dlbm.equals("021")&&(zldwdm.equals("330802009001")||zldwdm.equals("3308020090010000000")))
            {
                System.out.println(dlbm+","+zldwdm+","+tbmj+","+xzdwmj+","+lxdwmj+","+kcdlmj+","+tbdlmj);
            }

            return Double.parseDouble(props.get("TBDLMJ").toString()) < 0 ? null : mdl;
        }).filter(mdl -> mdl != null);

        JavaPairRDD<String, GeoItem> resultRDD = dltbMdlRDDall.mapToPair(mdl -> {
            String dlbm=mdl._2.getProp().get("DLBM").toString();

            String zldwdm=mdl._2.getProp().get("ZLDWDM").toString();
            String id = zldwdm + "##" + dlbm;
            Double area=Double.parseDouble(mdl._2.getProp().get("TBDLMJ").toString());
            GeoItem result = new GeoItem();
            result.setFid(id);
            Map<String, Object> props = new HashMap<>();
            props.put("DLBM", dlbm);
            props.put("ZLDWDM", zldwdm);
            props.put("AREA", area);
            result.setProp(props);

            return new Tuple2<>(id, result);
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        JavaPairRDD<String, GeoItem> xz1resultRDD = dltbXzdwRDD1.mapToPair(mdl -> {
            String dlbm=mdl._2.getProp().get("DLBM").toString();
            Double cd=Double.parseDouble(mdl._2.getProp().get("CD").toString());
            Double kd=Double.parseDouble(mdl._2.getProp().get("KD").toString());
            Double kcbl=Double.parseDouble(mdl._2.getProp().get("KCBL").toString());
            String zldwdm=mdl._2.getProp().get("KCTBDWDM1").toString();
            String id = zldwdm + "##" + dlbm;
            GeoItem result = new GeoItem();
            result.setFid(id);
            Map<String, Object> props = new HashMap<>();
            props.put("DLBM", dlbm);
            props.put("ZLDWDM", zldwdm);
            props.put("AREA", cd*kd*kcbl);
            result.setProp(props);

            return new Tuple2<>(id, result);
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        JavaPairRDD<String, GeoItem> xz2resultRDD = dltbXzdwRDD2.mapToPair(mdl -> {
            String dlbm=mdl._2.getProp().get("DLBM").toString();
            Double cd=Double.parseDouble(mdl._2.getProp().get("CD").toString());
            Double kd=Double.parseDouble(mdl._2.getProp().get("KD").toString());
            Double kcbl=Double.parseDouble(mdl._2.getProp().get("KCBL").toString());
            String zldwdm=mdl._2.getProp().get("KCTBDWDM2").toString();
            if(kcbl==0.5){
                String id = zldwdm + "##" + dlbm;
                GeoItem result = new GeoItem();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("DLBM", dlbm);
                props.put("ZLDWDM", zldwdm);
                props.put("AREA", cd*kd*kcbl);
                result.setProp(props);

                return new Tuple2<>(id, result);
            }
            return new Tuple2<>("EMPTY", null);
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        JavaPairRDD<String, GeoItem> lxresultRDD = dltbLxdwRDD.mapToPair(mdl -> {
            String dlbm=mdl._2.getProp().get("DLBM").toString();
            Double area=Double.parseDouble(mdl._2.getProp().get("MJ").toString());

            String zldwdm=mdl._2.getProp().get("ZLDWDM").toString();
            String id = zldwdm + "##" + dlbm;
            GeoItem result = new GeoItem();
            result.setFid(id);
            Map<String, Object> props = new HashMap<>();
            props.put("DLBM", dlbm);
            props.put("ZLDWDM", zldwdm);
            props.put("AREA", area);
            result.setProp(props);

            return new Tuple2<>(id, result);
        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        JavaPairRDD<String, GeoItem> kcresultRDD = dltbMdlRDD.mapToPair(mdl -> {
            String kcdlbm=mdl._2.getProp().get("KCDLBM").toString();
            if(!kcdlbm.equals(null) && kcdlbm.length()>=1){
                String zldwdm=mdl._2.getProp().get("ZLDWDM").toString();
                String id = zldwdm + "##" + kcdlbm;
                Double area=Double.parseDouble(mdl._2.getProp().get("KCDLMJ").toString());
                GeoItem result = new GeoItem();
                result.setFid(id);
                Map<String, Object> props = new HashMap<>();
                props.put("DLBM", kcdlbm);
                props.put("ZLDWDM", zldwdm);
                props.put("AREA", area);
                result.setProp(props);

                return new Tuple2<>(id, result);
            }
            return new Tuple2<>("EMPTY", null);

        }).filter(new Function<Tuple2<String, GeoItem>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, GeoItem> input) throws Exception {
                return !input._1.equals("EMPTY");
            }
        });

        Map<String, GeoItem> results = resultRDD.union(kcresultRDD).union(lxresultRDD).union(xz1resultRDD).union(xz2resultRDD).reduceByKey((m1, m2) -> {//按地类编码进行面积加和
            Double area = (Double) m1.getProp().get("AREA") + (Double) m2.getProp().get("AREA");
            GeoItem result = new GeoItem();
            String id=m1.getFid();
            result.setFid(id);
            Map<String, Object> props = new HashMap<>();
            props.put("AREA", area);
            result.setProp(props);
            return result;
        }).collectAsMap();
        return results;


    }
}
